package com.demo.kafka.client.sub_pub;

import com.demo.kafka.StartProperties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class SubPubProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        try (Producer<String, String> producer =
                     new KafkaProducer<>(StartProperties.getDefaultProperties("consumergroup"));) {

            for (int i = 0; i < 100; i++) {
                String message = "message-" + i;
                // producer采用异步批量的方式来发送消息，send方法会立即返回。
                Future<RecordMetadata> resultFuture = producer.send(new ProducerRecord<String, String>("test-partition-3", Integer.toString(i), message));

                // 如果你想要同步阻塞等待结果
                RecordMetadata rm = resultFuture.get();

                System.out.println("发送：" + message + " hasOffset: " + rm.hasOffset() + " partition: " + rm.partition() + " offset: " + rm.offset());
                TimeUnit.SECONDS.sleep(1L);
            }
        }
    }
}
